package work

import (
	"fmt"
	"log"
	"sync"
	"time"
)

/*
	功能：
	1、job-worker模式
	2、收到退出信号时，必须做完所有任务才能退出

	收获：
	1、go语言没有判断chan是否退出的功能
	2、判断某个chan是否退出，不应该直接对 channel 本身判断是否 close
	3、使用无缓存chan的同步特点，和close之后还会继续输出零值来辅助判断
	4、当同步阻塞就是未关闭，当输出就是close
*/

var (
	maxQueueSize int         = 5000
	maxWorkers   int         = 5
	Dispatcher   *dispatcher //调度器
	wg           sync.WaitGroup
	quitChan     chan struct{}
)

type job struct {
	CreateTime time.Time
	MyJob      string
}

type worker struct {
	id         int
	jobQueue   chan job
	workerPool chan chan job
}

func newWorker(id int, workerPool chan chan job) worker {
	return worker{
		id:         id,
		jobQueue:   make(chan job),
		workerPool: workerPool,
	}
}

func (w worker) start() {
	go func() {
		for {
			w.workerPool <- w.jobQueue

			job := <-w.jobQueue
			wg.Add(1)
			w.dojob(job)
		}
	}()
}

func (w *worker) dojob(j job) {
	defer wg.Done()
	log.Printf("worker:%d is start\n", w.id)
	time.Sleep(2 * time.Second)
	fmt.Println(j.MyJob)
	log.Printf("worker:%d is end\n", w.id)
}

func init() {
	Dispatcher = newDispatcher(maxQueueSize, maxWorkers)
	quitChan = make(chan struct{})

}

//调度器
type dispatcher struct {
	workerPool chan chan job //worker池
	maxWorkers int           //最大worker数量
	jobQueue   chan job      //任务队列
}

func newDispatcher(maxQueueSize, maxWorkers int) *dispatcher {
	return &dispatcher{
		jobQueue:   make(chan job, maxQueueSize),
		workerPool: make(chan chan job, maxWorkers),
		maxWorkers: maxWorkers,
	}
}

func (d *dispatcher) AddJob(jobInfo string) bool {
	job := job{CreateTime: time.Now(), MyJob: jobInfo}
	if ok := cancelled(); ok { //退出了，不能再请求数据了
		return false
	}
	d.jobQueue <- job
	return true
}

//辅助判断是否退出，也可以代替某个chan关闭
func cancelled() bool {
	select {
	case <-quitChan:
		return true
	default:
		return false
	}
}

func (d *dispatcher) OnStart() {
	for i := 0; i < d.maxWorkers; i++ {
		worker := newWorker(i+1, d.workerPool)
		worker.start()
	}

	go d.dispatch()
}

func (d *dispatcher) dispatch() {
	for j := range d.jobQueue {
		fmt.Println("---dispatch---len:", len(d.jobQueue))

		go func(j job) {
			workerJobQueue := <-d.workerPool
			workerJobQueue <- j
		}(j)
	}
}

func (d *dispatcher) OnStop() {
	close(quitChan)
	wg.Wait()
}
